package com.yanggu.kafka.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

@Component
@Configuration
public class KafkaConsumer {

    /**
     * 消费指定topic的数据
     */
    //@KafkaListener(topics = {"first"})
    //public void onMessage(ConsumerRecord<?, ?> record) {
    //    System.out.println(record);
    //}

    /**
     * 指定topic、partition、offset消费
     * 监听first主题的0号分区，从偏移量0开始
     */
    // 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
    //@KafkaListener(topicPartitions = {
    //        @TopicPartition(topic = "first", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0"))
    //})
    //public void onMessage2(ConsumerRecord<?, ?> record) {
    //    System.out.println(record);
    //}

    /**
     * 配置消息异常后的处理器
     */
    //@Bean
    //public ConsumerAwareListenerErrorHandler consumerAwareListenerErrorHandler() {
    //    return (message, exception, consumer) -> {
    //        System.out.println("消费异常：" + message.getPayload());
    //        return null;
    //    };
    //}

    /**
     * 消费消息时，发生异常指定异常处理器
     */
    //@KafkaListener(errorHandler = "consumerAwareListenerErrorHandler", topics = {"second"})
    //public void onMessage3() throws Exception {
    //    throw new Exception("简单消费-模拟异常");
    //}

    /**
     * 消息过滤器可以在消息抵达consumer之前被拦截，在实际应用中，我们可以根据自己的业务逻辑，筛选出需要的信息再交由KafkaListener处理，不需要的消息则过滤掉。
     * 配置消息过滤只需要为 监听器工厂 配置一个RecordFilterStrategy（消息过滤策略），返回true的时候消息将会被抛弃，返回false时，消息能正常抵达监听容器。
     *
     * @param consumerFactory
     * @return
     */
    //@Bean
    //public ConcurrentKafkaListenerContainerFactory filterContainerFactory(ConsumerFactory consumerFactory) {
    //    ConcurrentKafkaListenerContainerFactory<?, ?> factory = new ConcurrentKafkaListenerContainerFactory<>();
    //    factory.setConsumerFactory(consumerFactory);
    //    //被过滤的消息将被丢弃
    //    factory.setAckDiscarded(true);
    //    //返回true的消息会被过滤掉
    //    factory.setRecordFilterStrategy(consumerRecord -> consumerRecord.offset() % 2 == 0);
    //    return factory;
    //}

    /**
     * 这里只有offset为奇数的会被消费
     * containerFactory指定配置的filterContainerFactory
     */
    //@KafkaListener(topics = {"first"}, containerFactory = "filterContainerFactory")
    //public void onMessage4(ConsumerRecord<?, ?> record) {
    //    System.out.println(record);
    //}

    /**
     * 消息转发
     * 从first接收到消息, 然后转发到second
     */
    @KafkaListener(topics = {"first"})
    @SendTo({"second"})
    public String onMessage7(ConsumerRecord<?, ?> record) {
        return record.value() + "-forward message";
    }


}
